import pandas as pd
import pymysql
import time
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta


# 设置panda显示参数
pd.set_option("display.max_rows", 20)
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 220)

mysql_config = {
    'host': 'rm-wz92r8qt89c31fn40oo.mysql.rds.aliyuncs.com',
    'port': 3306,
    'user': 'bi_rw',
    'password': 'Xjz@&283*-)',
}


class DatabaseConnector:
    def __init__(self, db_type, config):
        self.db_type = db_type
        self.config = config
        if db_type == 'presto':
            self.conn = dbapi.connect(**config)
        elif db_type == 'hive':
            self.conn = hive.connect(**config)
        elif db_type == 'mysql':
            self.conn = pymysql.connect(**config)
        self.cursor = self.conn.cursor()

    def ddl(self, cmd):
        """
        执行ddl语句,
        :param cmd: ddl操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()

        # 使用presto的时候, 必须要加上fetchall()之后ddl语句才会执行, 不知道为什么
        if self.db_type == 'presto':
            self.cursor.fetchall()

        print(f"Finish running ddl: {cmd}")

    def dml(self, cmd):
        """
        执行增删改操作
            hive:   操作没有返回集, 使用fetchall会报错, 影响的行数也不可见
            presto: 结果集的类型是rows, insert语句的值是插入的行数, delete操作的值是None
        :param cmd: dml操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()
        if self.db_type == 'presto':
            infected_rows = self.cursor.fetchall()[0][0]
            print(f"Finish running dml ({infected_rows} rows infected): {cmd}")
        else:
            print(f"Finish running dml: {cmd}")

    def dql(self, cmd):
        """
        执行查询操作, 需要返回查询结果集的dataframe
        :param cmd: dql操作sql
        :return: dql结果
        """
        self.cursor.execute(cmd)
        dql_desc = self.cursor.description
        dql_rows = self.cursor.fetchall()
        print(f"Finish running dql: {cmd}")
        dql_df = None
        if len(dql_rows) != 0:
            columns = [col[0] for col in dql_desc]
            result = [dict(zip(columns, row)) for row in dql_rows]
            dql_df = pd.DataFrame(result)
            dql_df.columns = columns

        return dql_df


mysql_env = DatabaseConnector('mysql', mysql_config)



run_sql = """
insert into bi_all_in_one.tx_loan_after_data_tmp
with test1 as  -- 新增放款渠道关联 2025-05-13 新增风控标识 2025-07-01
(
select distinct 
        rrp.user_id
       ,rrp.order_no
       ,aau.assets_user_id 
from tx_credit_loan.r_repay_plan rrp
join tx_credit_loan.a_assets_user aau on rrp.user_id = aau.user_id  and  rrp.assets_code = aau.assets_code and rrp.product_code = aau.product_code  
)--  select count(user_id),count(distinct assets_user_id) ,count(assets_user_id) from test1;
,
test2 as 
(
select distinct flo.user_id 
	 ,flo.asset_code
	 ,flo.funding_order_no
	 ,flo.funding_loan_no 
	 ,aac.show_name as assets_name
	 ,loan_status 
from xxj_diversion.f_loan_order flo join xxj_diversion.a_assets_config aac on flo.asset_code = aac.assets_code 
where loan_status != 'N'
)
,
test3 as 
(
select t1.user_id
      ,t1.order_no
      ,t1.assets_user_id
      ,t2.asset_code
      ,t2.assets_name
      ,t2.loan_status
from test1 t1 
join test2 t2 on  t1.assets_user_id = cast(t2.user_id as char) and t1.order_no = t2.funding_order_no
) -- select count(1),count(user_id),count(assets_user_id ),count(distinct user_id) ,count(distinct assets_user_id ) from test3 ;
,
tmp as 
(
   select 
        t1.order_no  
       ,t1.user_id
       ,t1.loan_code  as funding_code
       ,t1.period
       ,t1.star_date  as star_date -- 开始日
       ,t1.pay_date   as due_date -- 到期日
       ,t1.principal 
       ,t1.finish_date
       ,t2.loan_date 
       ,t3.asset_code
       ,t2.product_code
--        ,min(t1.star_date) over (partition by t1.order_no,t1.user_id) as loan_date -- 资方28日还款日统一处理，故不使用当前逻辑获取放款日期
       ,sum(t1.principal) over (partition by t1.order_no,t1.user_id) as loan_amount
       ,count(t1.period)  over (partition by t1.order_no,t1.user_id) as periods
       ,sum(t1.actual_principal) as actual_principal
      -- ,max(substr(case when t1.actual_principal >= t1.principal then t1.finish_date else current_date() end,1,10)) as pay_date
    from tx_credit_loan.r_repay_plan t1 
    left join (select distinct user_id,assets_code,product_code,loan_code,order_no,order_status,cash_date as loan_date  from  tx_credit_loan.l_order ) t2 
         on t1.order_no = t2.order_no and t1.user_id = t2.user_id and t1.loan_code = t2.loan_code
    left join test3 t3 on t1.user_id = t3.user_id and t1.order_no = t3.order_no
    group by 1,2,3,4,5,6,7,8,9,10,11
    union all 
    select 
        frp.order_no  
       ,frp.user_id
       ,frp.funding_code 
       ,frp.periods  as period
       ,frp.star_date  as star_date -- 开始日
       ,frp.pay_date   as due_date -- 到期日
       ,frp.principal 
       ,frp.finish_date
       ,t2.loan_date 
       ,t2.asset_code
       ,'TB_ZB' product_code  -- 虚拟风控标识-标定当前code值走华章风控
--        ,min(frp.star_date) over (partition by frp.order_no,frp.user_id) as loan_date 
       ,sum(frp.principal) over (partition by frp.order_no,frp.user_id) as loan_amount
       ,count(frp.periods)  over (partition by frp.order_no,frp.user_id) as periods
       ,sum(frp.actual_principal) as actual_principal
    --   ,max(substr(case when frp.actual_principal >= frp.principal then frp.finish_date else current_date() end,1,10)) as pay_date
       from xxj_diversion.f_repay_plan frp 
       left join (select distinct user_id,asset_code,product_code,funding_code,order_no,loan_status,substr(loan_time,1,10)  as loan_date  from  xxj_diversion.f_loan_order flo where funding_code =  'JLD_LHYH') t2 
            on frp.funding_code = t2.funding_code and frp.order_no = t2.order_no and frp.user_id = t2.user_id
       where frp.funding_code = 'JLD_LHYH'
    group by 1,2,3,4,5,6,7,8,9,10,11
)-- select * from  tmp where due_date = '2025-03-18';
,
overdue_days_info  as  -- overdue_days 当前表内字段数据错误 
(
select user_id
      ,order_no
      ,period
      ,loan_code as funding_code 
      ,if(finish_date is null , 1 ,0 ) as b_tag_settle -- 是否结清
      ,datediff(coalesce(finish_date,date '{check_date}'),pay_date) as overdue_days2 
from tx_credit_loan.r_repay_plan 
where pay_date <= date '{check_date}'
 -- and (finish_date >= pay_date or finish_date is null  ) 
union all
 select 
       user_id
      ,order_no
      ,periods as period
      ,funding_code
      ,if(finish_date is null , 1 ,0 ) as b_tag_settle -- 是否结清
      ,datediff(coalesce(finish_date,date '{check_date}'),pay_date) as overdue_days2 
 from xxj_diversion.f_repay_plan 
 where funding_code = 'JLD_LHYH'
  and pay_date <= date '{check_date}'
  -- and (finish_date >= pay_date or finish_date is null) --
)-- select * from overdue_days_info;
,
order_overdue_days_goal as  --  已到期的订单的当前最大逾期天数
(
select 
    user_id 
   ,order_no
   ,funding_code 
   ,max(overdue_days2) as max_overdue_days2  -- 历史最大逾期天数
   ,max(if(b_tag_settle = 1 , overdue_days2 ,0)) as curr_overdue_days2   -- 当前最大逾期天数
from overdue_days_info  
group by 1,2,3
)
select t1.*,t2.max_overdue_days2,t2.curr_overdue_days2, date '{check_date}'  as check_date
from  tmp t1
left join order_overdue_days_goal t2 on t1.user_id = t2.user_id and t1.order_no = t2.order_no and t1.funding_code = t2.funding_code
where t1.due_date <= date '{check_date}' 
"""

# 循环开始日期 结束日期
start_date = datetime.strptime('2025-06-01', '%Y-%m-%d') 
end_date = datetime.strptime('2025-06-30', '%Y-%m-%d')

end_date_str = end_date.strftime('%Y-%m-%d')

while start_date <= end_date:
    start_date_str = start_date.strftime('%Y-%m-%d')

    params = {
        'check_date': start_date_str,
        # 'month_1' : 1,
        # 'month_2' : 2,
        # 'month_3' : 3,
        # 'c_ds'    :'2024-04-11'
    }


    mysql_env.dml(run_sql.format(**params)) # 执行

    #print(run_sql.format(**params)) # 检查打印循环命令是否传如参数正确
    
    start_date += timedelta(days=1)

